"""
Case Type   : Compatibility
Case Name   : 指定replace参数前撤销delete权限,导入合法数据，普通用户导入数据
Create At   : 2024/10/18
Owner       : @zhanghuan96
Description :
        1.创建数据文件
        2.创建兼容B库
        3.建用户
        4.设置参数
        5.将普通用户加入gs_role_copy_files群组
        6.进入普通用户创建表并撤销用户的delete权限
        6.截断表
        7.导入合法数据时指定参数replace
        8.查询表
        9.清理环境
Expect      :
        1.创建成功
        2.成功
        3.成功
        4.成功
        5.成功
        6.成功
        7.成功
        8.成功
        9.成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class LoadData0084(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.common = Common()
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.pri_u = Node('PrimaryDbUser')
        self.db_name = 'db_load0084'
        self.tb_name = 'tb_load0084'
        self.u_name = 'u_load0084'
        self.file_path = os.path.join(macro.DB_BACKUP_PATH,
                                      'load_data0084.txt')

    def test_load(self):
        text1 = '----step1:创建数据文件; expect:创建成功----'
        self.log.info(text1)
        echo_cmd = f'''rm -rf {self.file_path};
                    echo "1001+123+绝对路径" >> {self.file_path};
                    echo "1002+456+少一层目录层级" >> {self.file_path};
                    echo "1003+147+gfadfr" >> {self.file_path};
                    echo "1001+123+绝对路径" >> {self.file_path};
                    echo "1002+456+少一层目录层级" >> {self.file_path};
                    '''
        self.log.info(echo_cmd)
        res = self.common.get_sh_result(self.pri_sh.node, echo_cmd)
        self.log.info(res)
        self.assertEqual('', res, '文件写入失败')

        text2 = '----step2:创建兼容B库; expect:创建成功----'
        self.log.info(text2)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} dbcompatibility ='B';")

        self.log.info(sql_cmd)
        sql_res = self.pri_sh.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('CREATE DATABASE', sql_res, '兼容B库创建失败')

        self.log.info('初始用户加载B库')
        sql_cmd = f"select version();"
        self.log.info(sql_cmd)
        sql_res = self.pri_sh.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)

        text = '----step3:建用户和schema; expect:成功----'
        self.log.info(text)
        sql_cmd1 = (f'''set search_path = public;
        drop user if exists {self.u_name} cascade;
        create user {self.u_name} password '{macro.COMMON_PASSWD}';''')
        self.log.info(sql_cmd1)
        res = self.pri_sh.execut_db_sql(sql_cmd1,
                                        sql_type=f'-U {self.pri_u.db_user} '
                                                 f'-W {self.pri_u.db_password}',
                                        dbname=self.db_name)
        self.log.info(res)
        self.assertIn('CREATE ROLE', res, '查询失败')

        text = '----step4:设置参数; expect:成功----'
        self.log.info(text)
        msg = self.pri_sh.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_copy_server_files=on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.pri_sh.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            'enable_mot_server=on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.pri_sh.restart_db_cluster()
        self.log.info(msg)
        status = self.pri_sh.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '----step5:将普通用户加入gs_role_copy_files群组; expect：成功----'
        self.log.info(text)
        sql_cmd3 = (f"set search_path =  {self.u_name};"
                    f"grant gs_role_copy_files to {self.u_name};")
        self.log.info(sql_cmd3)
        sql_res3 = self.pri_sh.execut_db_sql(sql_cmd3, dbname=self.db_name)
        self.log.info(sql_res3)
        self.assertIn('GRANT ROLE', sql_res3)

        text = '----step6:进入普通用户创建表并撤销用户的delete权限; expect：成功----'
        self.log.info(text)
        sql_cmd3 = f'''set search_path =  {self.u_name};
        set b_compatibility_user_host_auth = on;
        set session authorization {self.u_name} password \
        '{macro.COMMON_PASSWD}';
        drop table if exists {self.tb_name};
        create table {self.tb_name}(
        c_01 int primary key,
        c_02 number,
        c_03 char(50));
        reset session authorization;
        revoke delete on table {self.tb_name} from {self.u_name};
        show grants for '{self.u_name}';'''
        self.log.info(sql_cmd3)
        res3 = self.pri_sh.execut_db_sql(sql_cmd3, dbname=self.db_name)
        self.log.info(res3)
        self.assertIn('CREATE TABLE', res3)
        self.assertIn('RESET', res3)
        self.assertIn('REVOKE', res3)
        self.assertIn('GRANT INSERT, SELECT, UPDATE, TRUNCATE, '
                      'REFERENCES, TRIGGER ON TABLE tb_load0084 TO u_load0084',
                      res3)

        text = '----step7:导入合法数据时指定参数replace; expect：成功----'
        self.log.info(text)
        sql_cmd2 = (f"set search_path =  {self.u_name};"
                    f"set session authorization {self.u_name} password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"load data infile '{self.file_path}' replace into table "
                    f"{self.tb_name} fields terminated by '+' ;")
        self.log.info(sql_cmd2)
        res2 = self.pri_sh.execut_db_sql(sql_cmd2,
                                         dbname=self.db_name)
        self.log.info(res2)
        self.assertIn('COPY 5', res2, '执行失败' + text)

        text = '----step8:查询表; expect：成功----'
        self.log.info(text)
        sql_cmd3 = f"select * from {self.u_name}.{self.tb_name};"
        self.log.info(sql_cmd3)
        sql_res3 = self.pri_sh.execut_db_sql(sql_cmd3, dbname=self.db_name)
        self.log.info(sql_res3)
        rows, _ = self.common.parse_sql_query_res(sql_res3)
        expect = [['1003', '147', 'gfadfr'],
                  ['1001', '123', '绝对路径'],
                  ['1002', '456', '少一层目录层级']]
        self.assertEqual(rows, expect, '执行失败' + text)

    def tearDown(self):
        text = '----step9:清理环境;expect:成功----'
        self.log.info(text)
        sql_cmd1 = f"drop user if exists {self.u_name} cascade;"
        self.log.info(sql_cmd1)
        sql_res1 = self.pri_sh.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)

        sql_cmd2 = f"drop database if exists {self.db_name};"
        self.log.info(sql_cmd2)
        sql_res2 = self.pri_sh.execut_db_sql(sql_cmd2, dbname='postgres')
        self.log.info(sql_res2)

        rm_cmd = f'''rm -rf {self.file_path};'''
        self.log.info(rm_cmd)
        res = self.common.get_sh_result(self.pri_sh.node, rm_cmd)
        self.log.info(res)

        guc_cmd = self.pri_sh.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            'enable_copy_server_files=off')
        self.log.info(guc_cmd)
        guc_cmd = self.pri_sh.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            'enable_mot_server=off')
        self.log.info(guc_cmd)
        text = '-----checkpoint;expect:成功-----'
        self.log.info(text)
        sql_cmd6 = f"checkpoint;"
        self.log.info(sql_cmd6)
        sql_res6 = self.pri_sh.execut_db_sql(sql_cmd6, dbname='postgres')
        self.log.info(sql_res6)
        self.assertIn('CHECKPOINT', sql_res6)
        msg = self.pri_sh.restart_db_cluster()
        self.log.info(msg)
        status = self.pri_sh.get_db_cluster_status(param='detail')
        self.log.info(status)
        self.assertTrue(status.count("Degraded") + status.count("Normal") == self.pri_sh.get_node_num() + 1)
        self.assertIn('DROP DATABASE', sql_res2)
        self.assertIn('DROP ROLE', sql_res1)
        self.log.info(f'----{os.path.basename(__file__)} end----')
